use std::sync::Arc;

use futures::{future::FutureExt, pin_mut, select};
use log::error;
use tokio::{spawn, sync};

use crate::{endpoint, Error, Result};

pub use crate::message::*;

/// 创建客户端通信代理.
///
/// uri为zeromq的endpoint字符串.
/// 返回消息收发器.
pub async fn bind(uri: &str) -> Result<(Sender, Reciever)> {
    endpoint::listen(uri).await.map(|(sender, reciever)| {
        (
            Sender {
                sender: sender.clone(),
            },
            Reciever { reciever, sender },
        )
    })
}

/// 客户端消息发送器.
#[derive(Clone)]
pub struct Sender {
    sender: endpoint::Sender,
}

impl Sender {
    /// 向客户端发送消息.
    pub async fn send(&mut self, conn_id: ConnectionId, rep: DownMessage) -> Result<()> {
        self.sender
            .send(conn_id, rep.to_bytes()?)
            .await
            .map_err(Error::from)
    }
}

/// 客户端消息接收器.
///
/// 同时检查客户端是否存在最近访问,否则删除与客户端连接.
pub struct Reciever {
    reciever: endpoint::Reciever,
    sender: endpoint::Sender,
}

impl Reciever {
    /// 从客户端接收消息.
    ///
    /// 返回(Self,已丢弃的连接id,(连接id,客户端请求)).
    pub async fn recv(
        mut self,
    ) -> Result<(Self, Vec<ConnectionId>, Option<(ConnectionId, UpMessage)>)> {
        let (conn_id, payload) = self.reciever.recv().await.map_err(Error::from)?;

        if let Ok(req) = UpMessage::from_bytes(&payload[..]) {
            Ok((self, Vec::new(), Some((conn_id, req))))
        } else {
            // 通知关闭连接
            self.sender.send(conn_id.clone(), Vec::new()).await.ok();
            Ok((self, vec![conn_id], None))
        }
    }

    pub fn split(self) -> (MainReciever, RuntimeReciever) {
        let (h_tx, h_rx) = sync::oneshot::channel();
        let h_tx = Arc::new(h_tx);

        let (m_tx, m_rx) = sync::mpsc::channel(8);
        let (r_tx, r_rx) = sync::mpsc::channel(8);

        spawn(async move {
            let recv_fut = self.recv().fuse();
            let stop_fut = h_rx.fuse();
            pin_mut!(recv_fut, stop_fut);

            loop {
                select! {
                    recv = recv_fut => {
                        match recv {
                            Ok((rx, timeout_clients, msg)) => {
                                if let Some((conn_id, msg)) = msg {
                                    match msg {
                                        UpMessage::VehicleControl(msg) => {
                                            if r_tx.send((conn_id, UpMessage::VehicleControl(msg))).await.is_err()
                                            {
                                                break;
                                            }

                                            if !timeout_clients.is_empty() && m_tx.send((timeout_clients, None)).await.is_err()
                                            {
                                                    break;
                                            }
                                        }

                                        _ => {
                                            if m_tx.send((timeout_clients, Some((conn_id, msg)))).await.is_err()
                                            {
                                                break;
                                            }
                                        }
                                    }
                                } else if m_tx.send((timeout_clients, None)).await.is_err() {
                                    break;
                                }


                                recv_fut.set(rx.recv().fuse());
                            }

                            Err(e) => {
                                error!("{}", e);
                                break;
                            }
                        }
                    }

                    _ = stop_fut => {
                        break;
                    }
                }
            }
        });

        (
            MainReciever {
                rx: m_rx,
                _h_tx: h_tx.clone(),
            },
            RuntimeReciever {
                rx: r_rx,
                _h_tx: h_tx,
            },
        )
    }
}

pub struct MainReciever {
    rx: sync::mpsc::Receiver<(Vec<ConnectionId>, Option<(ConnectionId, UpMessage)>)>,
    _h_tx: Arc<sync::oneshot::Sender<()>>,
}

impl MainReciever {
    pub async fn recv(
        mut self,
    ) -> Result<(Self, Vec<ConnectionId>, Option<(ConnectionId, UpMessage)>)> {
        if let Some(msg) = self.rx.recv().await {
            Ok((self, msg.0, msg.1))
        } else {
            Err(Error::new("Reciever failed".to_string()))
        }
    }
}

// TODO: 在runtime_server判断超时
pub struct RuntimeReciever {
    rx: sync::mpsc::Receiver<(ConnectionId, UpMessage)>,
    _h_tx: Arc<sync::oneshot::Sender<()>>,
}

impl RuntimeReciever {
    pub async fn recv(mut self) -> Result<(Self, ConnectionId, UpMessage)> {
        if let Some(msg) = self.rx.recv().await {
            Ok((self, msg.0, msg.1))
        } else {
            Err(Error::new("Reciever failed".to_string()))
        }
    }
}

pub type ConnectionId = endpoint::ConnectionId;
